package com.beiding.dbrequest.common;

import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.AttributeKey;
import io.reactivex.Observable;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.java.Log;

import java.io.Serializable;
import java.util.Map;

/**
 * 所有的请求都必须有响应
 * 正常,发送请求失败,连接失败等
 *
 * @author 丁常磊
 * @date 2021/7/16 12:42
 */
@Log
public class Terminal {

    /**
     * 消息分发器
     */
    @Getter
    @Setter
    protected Dispatcher dispatcher = new Dispatcher();

    /**
     * 拆包粘包长度单位,4个字节已经容纳int的最大值,通常也不会有这么大的包
     */
    @Getter
    @Setter
    protected Integer lengthFieldLength = 4;

    /**
     * 请求表
     */
    @Getter
    @Setter
    private RequestTable requestTable = new RequestTable();

    /**
     * 使用指定通道发起请企图
     *
     * @param ctx
     * @param uri
     * @param body
     * @return
     */
    public static Observable<Serializable> requestWithChannel(ChannelHandlerContext ctx, String uri, Serializable body) {
        Terminal terminal = getTerminal(ctx);
        return terminal.request(ctx, Message.request(uri, body));
    }

    /**
     * 静态方法
     *
     * @param ctx
     * @param message
     * @return
     */
    public static Observable<Serializable> requestWithChannel(ChannelHandlerContext ctx, Message message) {
        Terminal terminal = getTerminal(ctx);
        return terminal.request(ctx, message);
    }


    /**
     * 使用特定通道广播数据
     *
     * @param channelHandlerContext
     * @param message
     */
    public static Observable<?> broadcastWithChannel(ChannelHandlerContext channelHandlerContext, Message message) {
        message.setType(Message.TYPE_BROADCAST);
        Long id = Common.createId();
        message.setRequestId(id);
        return Observable.create(emitter -> {
            channelHandlerContext.writeAndFlush(message).addListener(future -> {
                if (future.isSuccess()) {
                    emitter.onComplete();
                }else  {
                    emitter.onError(future.cause());
                }
            });
        });
    }


    /**
     * 使用特定通道发起请求返回promise
     *
     * @param message
     * @return
     */
    public Observable<Serializable> request(ChannelHandlerContext channel, Message message) {
        return Observable.create(subscriber -> {
            request(channel, message, new ObservableResponseCallback(subscriber));
        });
    }

    /**
     * 使用特定通道发送数据
     *
     * @param channelHandlerContext
     * @param message
     * @param responseCallback
     */
    public void request(ChannelHandlerContext channelHandlerContext, Message message, ResponseCallback responseCallback) {

        message.setType(Message.TYPE_REQUEST);


        //设置请求id
        Long id = Common.createId();
        message.setRequestId(id);
        //构建请求器
        RequestLaunch requestLaunch = new RequestLaunch();

        //请求发起时间
        requestLaunch.setLaunchTime(System.currentTimeMillis());

        //设置回调函数
        requestLaunch.setResponseCallback(responseCallback);

        //设置请求时消息
        requestLaunch.setRequestMessage(message);

        //放入请求注册表 可超时
        this.requestTable.put(requestLaunch);

        //检查发送失败
        channelHandlerContext.writeAndFlush(message).addListener((ChannelFutureListener) future -> {
            if (!future.isSuccess()) {
                log.info("发送请求时,数据写入失败");
                Throwable cause = future.cause();
                if (cause != null) {
                    cause.printStackTrace();
                }
                requestTable.onResponse(Message.responseFor(message).msg(future.cause().getMessage()).status(Message.STATUS_INVOKE_ERROR));
            }
        });

    }

    /**
     * 添加通道激活时的行为
     */
    public static void addActiveCallback(ChannelHandlerContext ctx,ChannelCallback callback) {
        ctx.channel().attr(CHANNEL_CALLBACK_KEY).get().addActive(callback);
    }


    /**
     * 添加关闭行为
     * @param ctx
     * @param callback
     */
    public static void addInactiveCallback(ChannelHandlerContext ctx,ChannelCallback callback) {
        ctx.channel().attr(CHANNEL_CALLBACK_KEY).get().addInactive(callback);
    }

    /**
     * 添加通道激活时的行为
     */
    public static void removeActiveCallback(ChannelHandlerContext ctx,ChannelCallback callback) {
        ctx.channel().attr(CHANNEL_CALLBACK_KEY).get().removeActive(callback);
    }


    /**
     * 添加关闭行为
     * @param ctx
     * @param callback
     */
    public static void removeInactiveCallback(ChannelHandlerContext ctx,ChannelCallback callback) {
        ctx.channel().attr(CHANNEL_CALLBACK_KEY).get().removeActive(callback);
    }



    public class H extends ChannelInboundHandlerAdapter {

        /**
         * 覆盖通道建立声明周期方法
         * 该方法将记录当前通道对应的终端
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            //绑定该终端
            ctx.channel().attr(TERMINAL_KEY).set(Terminal.this);

            //绑定回调事件存储
            ChannelHolderForCallback channelHolderForCallback = new ChannelHolderForCallback();
            channelHolderForCallback.setChannelHandlerContext(ctx);
            ctx.channel().attr(CHANNEL_CALLBACK_KEY).set(channelHolderForCallback);

            channelHolderForCallback.active();

        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object o) throws Exception {
            Message msg = (Message) o;
            /*
            设置当前处理读请求的终端对象
             */
            HandlerCommon.onMessage(ctx, msg, dispatcher, requestTable);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {

            //通道被关闭
            ctx.channel().attr(CHANNEL_CALLBACK_KEY).get().inactive();

        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent event = (IdleStateEvent) evt;
                if (IdleState.READER_IDLE.equals((event.state()))) {
                    ctx.writeAndFlush(new Message().type(Message.TYPE_HEARTBEAT)).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                }
            }
            super.userEventTriggered(ctx, evt);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        }
    }


    protected void doRequest(ChannelHandlerContext channel, Message message, ResponseCallback responseCallback) {
        try {
            if (channel == null) {
                if (responseCallback != null) {
                    Message resp = Message.response(Message.STATUS_INVOKE_ERROR, "非可用通道", null);
                    responseCallback.onResponse(resp);
                }
            } else {
                request(channel, message, responseCallback);
            }
        } catch (Exception e) {
            if (responseCallback != null) {
                responseCallback.onResponse(Message.response(Message.STATUS_INVOKE_ERROR, e.getMessage(), null));
            }
        }
    }


    /**
     * 当前的终端实例
     */
    private static final AttributeKey<Terminal> TERMINAL_KEY = AttributeKey.newInstance("BIND_TERMINAL");

    private static final AttributeKey<ChannelHolderForCallback> CHANNEL_CALLBACK_KEY = AttributeKey.newInstance("CHANNEL_CALLBACK");

    public static Terminal getTerminal(ChannelHandlerContext ctx) {
        return ctx.channel().attr(TERMINAL_KEY).get();
    }


}
